package fast;

import java.io.*;
import java.util.*;

/**
 * SlaveProcessor Class
 * 
 * The Slave Processor is a node in a distributed system that
 * receives commands from a single Master Processor.
 * The commands relate to peering with the Master, Checking, Loading 
 * and Searching .iso image files and disconnecting from the distributed system.
 * 
 * @version 0.1
 * @author Nick Dyer
 * @date 12 April 2010
 */


public class SlaveProcessor extends Thread {
	private String filePath;
	private byte[] fileBytes;
	private Thread sTalker;
	private Thread sListen;
	private Thread sUDPListen;
	private Buffer sTalkBuffer;
	private Buffer sListenBuffer;
	public String consoleOut = "";
	public int fileStart;

	private final int TCP_TALK_PORT = 5000;
	private final int TCP_LISTEN_PORT = 10001;
	private final int UDP_LISTEN_PORT = 6000;
	private final int BUFFER_LEN = 50;
	private final int MAX_FILE_SIZE = 524288000;
	private boolean running = true;
	private boolean searching = false;
	private String masterIp = "";

	public SlaveProcessor() throws FileNotFoundException, IOException {

		FilePath fp = new FilePath();
		filePath = fp.getFilePath();

		System.out.println("path = " + fp.getFilePath());

		// prep some buffers, listeners, talkers
		this.sTalkBuffer = new Buffer(BUFFER_LEN);
		this.sListenBuffer = new Buffer(BUFFER_LEN);

		this.sTalker = new Thread(new TCPTalker(sTalkBuffer));
		this.sListen = new Thread(new TCPListener(TCP_LISTEN_PORT,
				sListenBuffer));
		this.sUDPListen = new Thread(new UDPListener(UDP_LISTEN_PORT,
				sListenBuffer));
	}
	
	/**
	 * Start the sTalker buffer, sListen buffer and sUDPListen buffer Threads.
	 * Remove commands from the buffer then process it using the decision making engine.
	 * Forward the data to the relevant method.
	 * 
	 */
	public void run() {
		// kick off the talker and listener threads
		this.sTalker.start();
		this.sListen.start();
		this.sUDPListen.start();

		// run until halted
		while (this.running) {
			try {
				boolean isMaster = false;

				// remove an entry from the listenBuffer
				String bufferString = sListenBuffer.remove();

				// split the string, token on ":". [0] = ip, [1], = id, [2] =
				// command, [3] = data

				String[] parsed = bufferString.split("\\|");

				String ip = parsed[0];
				String id = parsed[1];
				String command = parsed[2];
				String data = "";

				if (parsed.length > 3) {
					data = parsed[3];
				}

				if (masterIp.equals(ip)) {
					isMaster = true;
				}

				// All of these need a corresponding method
				// Decision maker based on received command
				if (command.equals(CommandsEnum.SYSTEM_PEER.toString())
						&& masterIp.equals("")) {
					sysPeer(ip, id, data);
				} else if (command.equals(CommandsEnum.SYSTEM_DONE.toString())
						&& isMaster) {
					sysDone(ip, id, data);
				} else if (command.equals(CommandsEnum.PROC_LOAD.toString())
						&& isMaster) {
					procLoad(data);
				} else if (command.equals(CommandsEnum.PROC_SEARCH.toString())
						&& isMaster) {
					procSearch(data);
				} else if (command.equals(CommandsEnum.SYSTEM_ERROR.toString())
						&& isMaster) {
					sysError(ip, id, data);
				} else if (command.equals(CommandsEnum.SYSTEM_EXIT.toString())
						&& isMaster) {
					sysExit();
				} else if (command.equals(CommandsEnum.SYSTEM_SHUTDOWN
						.toString())
						&& isMaster) {
					sysShut();
				} else {
					System.out.println(command + " Invalid command received!");
					// invalid command, do nothing
				}
			} catch (InterruptedException e) {
				System.out.println(e);
			}
		}
	}

	/**
	 * Prints "Done" plus the Master IP, ID number and data to the console
	 * on completion of the peering process.
	 * 
	 * @param 	ip	
	 * @param 	id
	 * @param	data	
	 */	
	private void sysDone(String ip, String id, String data) {
		consoleOut = "Done " + ip + " " + id + " " + data;
		System.out.print(consoleOut + "\n");
		// System.out.println("Done " + ip + " " + id + " " + data);
	}

	/**
	 * Deals with a Peer broadcast from a MP if the masterIp is not set.
	 * 
	 * @param  ip  	The IP Address of the Master Processor
	 * @param  id 	The ID of the command sent as specified by the MP
	 * @param  data !!Can't remember (Port number maybe)!!
	 */
	private void sysPeer(String ip, String id, String data) {
		// respond with a JOIN message
		// PEER is a broadcast from the master
		// System.out.println("Received Peer from Master Processor");

		consoleOut = "Received Peer from Master Processor";
		System.out.println(consoleOut);

		if (masterIp == "") {
			masterIp = ip;
			System.out.println("Master IP set to " + ip);

			try {
				sTalkBuffer.add(ip, TCP_TALK_PORT, Integer.valueOf(id),
						CommandsEnum.SYSTEM_JOIN.toString(), String
								.valueOf(TCP_LISTEN_PORT));
			} catch (NumberFormatException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	/**
	 * Executed by the "Leave" button in the SlaveUI.
	 * Stops any searches that are running and nulls the masterIP variable.
	 * The slave can be started again if a MP sends a Peer broadcast.
	 * 
	 * @param  id 	The ID of the command sent as specified by the MP
	 */
	public void sysExit() {
		System.out
				.println("Stopping all current activities, dropping master & sending a LEAVE");

		if(searching){
			searching = false;
		}
		
		if (!masterIp.equals("")) {
			try {
				sTalkBuffer.add(masterIp, TCP_TALK_PORT, 0, "LEAVE", null);
			} catch (NumberFormatException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			masterIp = "";
			fileBytes = null;
		}
	}

	/**
	 * Called by the "Quit" button in the SlaveUI.
	 * Runs the sysExit method, stops the slave from running, 
	 * then shuts down.
	 * 
	 * @param  id 	The ID of the command sent as specified by the MP
	 */
	public void sysShut() {
		sysExit();
		System.out.println("WARNING! Shutting down the slave processor.");
		running = false;
		System.exit(0);
	}

	/**
	 * Receives the data from the procLoad command, 
	 * parses data and extracts the filepath, startbyte and number of bytes.
	 * Performs file validation via the validateFile method.
	 * Reads in the file via the getFile method
	 * 
	 * @param  data 	Contains the Filepath, start byte and number of bytes to search.
	 */
	public void procLoad(String data) {
		System.out.println("Load Received " + data);
		// Parse the data
		String[] parse = data.split(",");

		// Extract the Data field and parse it further into filePath, start byte
		// and number of bytes to search
		String filePath = this.filePath + parse[0];
		String startByte = parse[1];
		String numByte = parse[2];

		// Convert the String values into byte values
		int sb = Integer.valueOf(startByte);
		int nb = Integer.valueOf(numByte);

		// Call validateFile to check the file is OK.
		if (validateFile(filePath)) {

			// If all is well with the file call getFile.
			try {
				getFile(filePath, sb, nb);
				System.out.println(filePath + " loaded.");
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	/**
	 * Called by procLoad to read in the required part of
	 * the file using the startByte and numBytes variables.
	 * 
	 * @param  filePath	Path to the target file
	 * @param  start	Start byte value
	 * @param  numBytes	The number of bytes that need to be read in.
	 */
	public void getFile(String filePath, int start, int numBytes)
			throws IOException {
		InputStream is = new FileInputStream(new File(filePath));

		// Initialise the fileBytes array
		// Maximum size of fileBytes[] I could initialise when loading using
		// java -xMx1024M is 577810752 bytes (551MB) so capping filesize per
		// node at
		// 500MB seems sensible.

		if ((numBytes) > MAX_FILE_SIZE) {
			throw new IOException();
		}

		fileBytes = new byte[numBytes];
		fileStart = start;

		// Start at the specified offset
		is.skip(start);

		int offset = 0;
		int numRead = 0;

		// is.read(fileBytes, 0, fileBytes.length);

		while (offset < (fileBytes.length)
				&& (numRead = is.read(fileBytes, offset, fileBytes.length
						- offset)) >= 0) {
			offset += numRead;
		}

		// Commented out to stop throwing the error for the last slave
		// this happens because the split file does overlap to cover searching
		// of the whole file.
		/*
		 * if (offset < fileBytes.length) { throw new IOException(filePath +
		 * " StartByte " + start + " EndByte " + (start + numBytes) + " " +
		 * " Could not read the entire part file."); }
		 */

		is.close();
	}

	/**
	 * Called by procLoad to check the target file exists
	 * is readable and of the correct type (.iso) by 
	 * checking the file extension.
	 * 
	 * @param  filePath 	The location and name of the file.
	 * @return	True if the file is readable, and a .iso image.
	 */
	public boolean validateFile(String filePath) {

		// Create the file object.
		File f = new File(filePath);

		// Check that the file exists and is readable.
		if (f.exists() && f.canRead()) {

			if (f.getName().endsWith(".iso") && f.length() <= Integer.MAX_VALUE) {
				System.out.println(filePath + " is a valid image type.");
				return true;
			} else
				System.out.println(filePath
						+ " is not of a valid type or size.");
		}
		return false;

	}

	/**
	 * This method contains the Boyer-Moore string searching algorithm.
	 * It received the search string and searched the already loaded file for matches.
	 * 
	 * Code here modified from the code found at the following URL
	 * http://en.literateprograms
	 * .org/Special:Downloadcode/Boyer-Moore_string_search_algorithm_%28Java%29
	 * 
	 * @param 	searchString
	 * @return	True when searching a file.
	 */
	public boolean procSearch(String searchString) {

		System.out.println("Search received " + searchString);
		// convert the string to an array of bytes.
		byte[] searchBytes = searchString.getBytes();

		searching = true;

		List<Integer> matches = new ArrayList<Integer>();
		int dataLen = fileBytes.length;
		int searchLen = searchBytes.length;

		Map<Character, Integer> indexTable = new HashMap<Character, Integer>();

		for (int i = searchLen - 1; i >= 0; i--) {
			char c = (char) searchBytes[i];
			if (!indexTable.containsKey(c))
				indexTable.put(c, i);
		}

		int alignedAt = 0;

		while (alignedAt + (searchLen - 1) < dataLen && searching) {

			for (int indexInPattern = searchLen - 1; indexInPattern >= 0; indexInPattern--) {
				int indexInText = alignedAt + indexInPattern;
				char x = (char) fileBytes[indexInText];
				char y = (char) searchBytes[indexInPattern];

				if (indexInText >= dataLen) {
					break;
				}

				if (x != y) {
					Integer r = indexTable.get(x);

					if (r == null) {
						alignedAt = indexInText + 1;
					}

					else {
						int shift = indexInText - (alignedAt + r);
						alignedAt += shift > 0 ? shift : 1;
					}

					break;
				} else if (indexInPattern == 0) {
					matches.add(fileStart + alignedAt);
					alignedAt++;
				}

			}
		}

		if (matches.isEmpty()) {
			System.out.println("No matches found for " + searchString);
		} else {
			System.out.println(matches.size() + " string matches found for "
					+ searchString);
		}

		for (int i = 0; i < matches.size(); i++) {
			System.out.println("Match at: " + matches.get(i));

			if ((matches.size() - i) > 10) {

				// send a packet with 10 results in
				String data = "";

				for (int j = 0; j < 10; j++) {
					data = data + Integer.toString(matches.get(i + j)) + ",";
				}

				try {
					this.sTalkBuffer.add(masterIp, TCP_TALK_PORT, 9,
							CommandsEnum.PROC_RESULT.toString(), data);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}

				// bump i by 9
				i += 9;

			} else {
				// send a packet with 1 result in
				try {
					this.sTalkBuffer.add(masterIp, TCP_TALK_PORT, 9,
							CommandsEnum.PROC_RESULT.toString(), Integer
									.toString(matches.get(i)));
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}

		searching = false;
		return true;
	}

	
	@Deprecated
	private void sysError(String ip, String id, String data) {
		System.out.println(ip + " " + id + " " + data);
	}


	// Required for local testing in Junit.
	public void setMaster(String ip) {
		this.masterIp = ip;
	}

}